package com.showdor.springboot.kafka;


import com.showdor.springboot.ConsumerRecordDAO;
import com.showdor.springboot.SimpleRecordDAOProxy;
import org.apache.kafka.clients.consumer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author showdor
 * @email chuanqi@outlook.com
 * @date 2023/7/5
 */
@Component
public class ConsumerHandler {
    private static final Logger logger = LoggerFactory.getLogger(ConsumerHandler.class);

    //kafka消费对象
    private KafkaConsumer<Object, Object> consumer;
    //线程池对象
    private ExecutorService executors;
    @Autowired
    KafkaConsumerConfig kafkaConsumerConfig;



    //kafka属性配置()
    public static Properties initConfig() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.139.116:32092");
        props.put("group.id", "test01");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1500);// 单次poll的数量,批量消费时配置
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //下面两个参数是我新加的，不加的情况下会报错
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "60000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "60000");
        return props;
    }


    /**
     * 多线程消费kafka数据
     */
    //@PostConstruct
    public void execute() throws InterruptedException {
        executors = new ThreadPoolExecutor(3, 5, 5000L, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
        while (true) {
            if (consumer != null) {
                //定义线程容器,通过java的引用类型记录数据。
                // Set<Thread> threadsContainer = new HashSet<>();
                //自定义的方法
                //setThreadFactory(executors, threadsContainer);

                ConsumerRecords<Object, Object> consumerRecords = consumer.poll(Duration.ofMillis(100));
                if (!consumerRecords.isEmpty()) {
                    for (final ConsumerRecord record : consumerRecords) {
                        //executors.submit(new OneWork(dao, record));
                        //commitOffset();
                    }
                    //executors.awaitTermination(3, TimeUnit.MICROSECONDS);

                    //方法二：
/*                    Thread mainThread = Thread.currentThread();
                    ThreadGroup mainThreadThreadGroup = mainThread.getThreadGroup();
                    //获取线程组中的线程。
                    int count = mainThreadThreadGroup.activeCount();
                    logger.info("count:" + count);
                    Thread[] threads = new Thread[count];
                    //enumerate 枚举，recurse 递归
                    mainThreadThreadGroup.enumerate(threads, true);
                    Stream.of(threads).filter(Thread::isAlive).forEach(thread -> logger.info("方法2：线程池的线程：" + thread));*/
                }

            } else {
                Properties props = initConfig();
                consumer = new KafkaConsumer<>(props);
                consumer.subscribe(Collections.singleton("netcross"));
            }
        }

    }

    private void commitOffset() {
        try {
            consumer.commitAsync();

        } catch (Exception E) {
            consumer.commitSync();
        }
    }

    private static void setThreadFactory(ExecutorService executorService, Set<Thread> threadsContainer) {
        if (executorService instanceof ThreadPoolExecutor) {
            ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executorService;
            //获取线程工厂
            ThreadFactory oldThreadFactory = threadPoolExecutor.getThreadFactory();
            //在把线程工程设置到包装类 DelegatingThreadFactory ，再设置回线程池。
            threadPoolExecutor.setThreadFactory(new MyThreadFactory(oldThreadFactory, threadsContainer));
        }

    }

    //我的的线程工厂
    private static class MyThreadFactory implements ThreadFactory {

        private final ThreadFactory threadFactory;
        private final Set<Thread> threadsContainer;

        private MyThreadFactory(ThreadFactory threadFactory, Set<Thread> threadsContainer) {
            this.threadFactory = threadFactory;
            this.threadsContainer = threadsContainer;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = threadFactory.newThread(r);
            //cache thread 记录线程
            threadsContainer.add(thread);
            //删除不存活的线程
//            threadsContainer.removeIf(next -> !next.isAlive());
            return thread;
        }
    }
}
